package com.qupeng.demo.kafka.kafkaspringbootconsumer.listener.single;

import com.qupeng.demo.kafka.kafkaspringbootconsumer.config.KafkaProperties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.support.Acknowledgment;

@Configuration
@ConditionalOnProperty(name = "app.kafka.consumer.mode.single-listener-container", havingValue = "true", matchIfMissing = false)
public class ListenerContainerConfig<K, V> {

    @Autowired
    private KafkaProperties kafkaProperties;

    @Bean
    public KafkaMessageListenerContainer<K, V> listenerContainer(ConsumerFactory<K, V> consumerFactory) {
        KafkaMessageListenerContainer<K, V> container = new KafkaMessageListenerContainer<>(consumerFactory, listenerContainerConfigs());
        container.setupMessageListener(messageListener());
        //container.start();
        return container;
    }

    @Bean
    public MessageListener<K, V> messageListener() {
        return new MessageListener<>() {
            @Override
            public void onMessage(ConsumerRecord<K, V> objectObjectConsumerRecord) {
                System.out.println(objectObjectConsumerRecord);
            }

            @Override
            public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
                MessageListener.super.onMessage(data, acknowledgment);
            }

            @Override
            public void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer) {
                MessageListener.super.onMessage(data, consumer);
            }

            @Override
            public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
                MessageListener.super.onMessage(data, acknowledgment, consumer);
            }
        };
    }

    @Bean
    public ContainerProperties listenerContainerConfigs() {
        ContainerProperties containerProps = new ContainerProperties(kafkaProperties.getTopics().toArray(new String[0]));
        containerProps.setGroupId(kafkaProperties.getGroupId());
        return containerProps;
    }
}
